package com.qisi.driver;

import com.qisi.functions.*;
import com.qisi.pojo.KafkaAlertLog;
import com.qisi.pojo.Rule;
import com.qisi.utils.Config;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SentryAlertJobDriver {
    public static void main(String[] args) throws Exception {

        // 全局配置
        Configuration config = Config.initApplicationConfig(args, SentryAlertJobDriver.class);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.setParallelism(2);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(4,3000L));
        env.enableCheckpointing(600000L);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.setStateBackend(new FsStateBackend("file:///qisi/gitlab/flink-real-time-alert/cp"));
//        env.setStateBackend(new FsStateBackend("hdfs://dev01:8020/flink/checkpoint/flink-real-time-alert"));
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.getCheckpointConfig().disableCheckpointing();


        // 捕获 MySQL 中的规则流
        DataStreamSource<Rule> mysqlRuleStream = env.addSource(SourceFunctions.getCustomMysqlSource(config));
        // 捕获 Mysql 中的维表
        // DataStreamSource<String> mysqlAppDim = env.addSource(SourceFunctions.getMysqlDimSource());

        // 获取 Kafka 日志流
        DataStreamSource<String> kafkaSource = env.addSource(SourceFunctions.getKafkaEventSource());

        // ETL 筛选出 MessageLog
        SingleOutputStreamOperator<String> msg = kafkaSource.flatMap(new LogBeanFlatMapFunction());

        // 连接 日志流 & 规则广播流
        ConnectedStreams<String, Rule> connectStream = msg.connect(mysqlRuleStream.broadcast());
        // 匹配出告警的 LogMessage
        SingleOutputStreamOperator<KafkaAlertLog> alertStream = connectStream.process(new RuleMatchBroadCastProcessFunction(config));

        // 发送告警信息到 AlertManager
        alertStream.addSink(new AlertManagerSinkFunction(config));

        // 告警服务发送到 Mysql 满足告警统计的需求
        alertStream.addSink(SinkFunctions.getMysqlAlertSinkFunction(config));

        // 告警结果发送到 Kafka 满足监控的需求
        alertStream.filter(KafkaAlertLog::isSentryAlertData).map(KafkaAlertLog::getJson).addSink(SinkFunctions.getFlinkKafkaProducer());

        alertStream.print();

        env.execute("alert");
    }

}
